package com.yang.common.amqp.rabbit;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.yang.common.amqp.MessageSender;
import com.yang.common.amqp.rabbit.constant.HeaderConstants;
import com.yang.common.core.util.MdcUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;

/**
 * @author: Billyang
 * @date: 2024/9/17 23:44
 */
@Primary
@Component
@Slf4j
public class RabbitMessageSender implements MessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void sendNormalMessage(String strategyName, Object data, String exchange, String routingKey) {
        sendMessage(strategyName, data, exchange, routingKey, null);
    }

    @Override
    public void sendFanoutMessage(Object data, String exchange) {
        sendMessage(null, data, exchange, null, null);
    }

    @Override
    public void sendDelayMessage(String strategyName, Object data, String exchange, String routingKey,
                                 Integer delaySeconds) {
        sendMessage(strategyName, data, exchange, routingKey, delaySeconds);
    }

    /**
     * 发送消息
     *
     * @param strategyName
     * @param data
     * @param exchange
     * @param routingKey
     * @param delaySeconds
     */
    private void sendMessage(String strategyName, Object data, String exchange, String routingKey,
                             Integer delaySeconds) {
        if (rabbitTemplate == null) {
            log.warn("rabbitTemplate not init");
            return;
        }

        String correlationId = RandomUtil.randomString(32);
        String paramsStr = JSON.toJSONString(data);
        rabbitTemplate.convertAndSend(exchange, routingKey, paramsStr, messageToSend -> {
            MessageProperties messageProperties = messageToSend.getMessageProperties();
            if (strategyName != null) {
                messageProperties.setHeader(HeaderConstants.HEADER_STRATEGY_NAME, strategyName);
            }
            messageProperties.setHeader(HeaderConstants.HEADER_PARAMS_CLASS, data.getClass().getName());
            messageProperties.setMessageId(MdcUtil.get());
            if (delaySeconds != null) {
                Integer delayMillis = delaySeconds * 1000;// 设置延时毫秒值
                messageProperties.setDelay(delayMillis);// x-delayed延时
                messageProperties.setExpiration(String.valueOf(delayMillis));// x-dead-letter延时
            }
            return messageToSend;
        }, new CorrelationData(correlationId));
        log.info("convertAndSend,correlationId:{},strategyName:{},toJson:{},exchange:{},routingKey:{},delaySeconds:{}",
                correlationId, strategyName, paramsStr, exchange, routingKey, delaySeconds);
    }
}
